package com.jaymin.flink.datastream.sink;

import com.jaymin.flink.datastream.transformation.Access;
import com.jaymin.flink.datastream.utils.MysqlUtils;
import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.PreparedStatement;

/**
 * @author jaymin
 * @since 2024/4/5 20:50
 */
public class PKMysqlSink extends RichSinkFunction<Tuple2<String, BigDecimal>> {
    Connection connection;
    PreparedStatement insertPrepareStatement;
    PreparedStatement updatePrepareStatement;

    @Override
    public void open(OpenContext openContext) throws Exception {
        super.open(openContext);

        connection = MysqlUtils.getConnection();
        insertPrepareStatement = connection.prepareStatement("insert into pk(domain,traffic) values (?,?)");
        updatePrepareStatement = connection.prepareStatement("update pk set traffic = ? where domain = ?");
    }

    @Override
    public void close() throws Exception {
        super.close();
        connection.close();
        insertPrepareStatement.close();
        updatePrepareStatement.close();
    }

    /**
     * 弊端，来一条就要执行一次
     * @param value The input record.
     * @param context Additional context about the input record.
     * @throws Exception
     */
    @Override
    public void invoke(Tuple2<String, BigDecimal> value, Context context) throws Exception {
        System.out.println("======invoke========" + value.f0 + "traffic:" + value.f1);
        updatePrepareStatement.setBigDecimal(1, value.f1);
        updatePrepareStatement.setString(2, value.f0);
        updatePrepareStatement.execute();
        if (updatePrepareStatement.getUpdateCount() == 0) {
            insertPrepareStatement.setString(1, value.f0);
            insertPrepareStatement.setBigDecimal(2, value.f1);
            insertPrepareStatement.execute();
        }
    }
}
